package org.spbsu.apco.client;

import org.perf4j.StopWatch;
import org.spbsu.apco.common.utils.StopWatchFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * User: ortemij
 * Date: 11.01.13
 * Time: 19:53
 */
public class ClientThreadPoolExecutor extends ThreadPoolExecutor {

    private static final int QUEUE_SIZE_PER_THREAD = 256;

    private final AtomicInteger taskCounter = new AtomicInteger();

    public ClientThreadPoolExecutor(int threadsCount) {
        super(threadsCount, threadsCount, 0, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(threadsCount * QUEUE_SIZE_PER_THREAD));
    }

    @Override
    public void execute(Runnable command) {
        taskCounter.incrementAndGet();
        super.execute(command);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
        stopWatch.stop("client__pool-thread-idle");
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        final int value = taskCounter.decrementAndGet();
        final StopWatch stopWatch = StopWatchFactory.createOrGetLocal();
        stopWatch.start();
        if (value < getCorePoolSize()) {
            synchronized (this) {
                this.notify();
            }
        }
    }
}
